package com.azh.server.datam.logger;

import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.Arrays;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import com.azh.server.util.ExceptionUtils;
import com.azh.server.util.Parser;
import com.google.common.primitives.Bytes;

/**
 * @author zhaozhenhang <zhaozhenhang@kuaishou.com>
 * Created on 2022-04-25
 * 日志文件
 *
 * 日志文件格式：[txCheckSum][log1][log2][tail]
 * txCheckSum 是后续所有log的checkSum，int类型
 *
 * 每条log 格式：[size][checkSum][data]
 * size 标示data长度， checkSum该日志的校验和
 */
public class LoggerImpl implements Logger{

    private static final int seed = 13331;

    private static final int of_size = 0;
    private static final int of_checksum = of_size + 4;
    private static final int of_data = of_checksum + 4;

    public static final String log_suffix = ".log";

    private RandomAccessFile file;
    private FileChannel channel;
    private Lock lock;

    // 当前日志 pointer 位置
    private long position;

    // 初始化时的记录，log 操作不更新
    private long fileSize;

    private int txCheckSum;

    public LoggerImpl(RandomAccessFile file, FileChannel channel) {
        this.file = file;
        this.channel = channel;
        this.lock = new ReentrantLock();
    }

    public LoggerImpl(RandomAccessFile file, FileChannel channel, int txCheckSum) {
        this.file = file;
        this.channel = channel;
        this.txCheckSum = txCheckSum;
        lock = new ReentrantLock();
    }

    /**
     *
     */
    public void init() {
        long size = 0;
        try {
            size = file.length();
        } catch (IOException e) {
            ExceptionUtils.getMessage(e);
        }
        if (size < 4) {
            ExceptionUtils.getMessage(new RuntimeException("Bad log file!"));
        }
        ByteBuffer buffer = ByteBuffer.allocate(4);
        try {
            channel.position(0);
            channel.read(buffer);
        } catch (IOException e) {
            ExceptionUtils.getMessage(e);
        }
        int txCheckSum = Parser.parseInt(buffer.array());
        this.fileSize = size;
        this.txCheckSum = txCheckSum;
        checkAndRemoveTail();
    }

    private int calCheckSum(int check, byte[] log) {
        for (byte b : log) {
            check = check * seed + b;
        }
        return check;
    }

    /**
     * 迭代器模式，不断从文件中读取下一条日志，然后把data解析返回
     * @return
     */
    private byte[] internNext() {
        if (position + of_data >= fileSize) {
            return null;
        }
        // 读取 size
        ByteBuffer buffer = ByteBuffer.allocate(4);
        try {
            channel.position(position);
            channel.read(buffer);
        } catch (IOException e) {
            ExceptionUtils.getMessage(e);
        }
        int size = Parser.parseInt(buffer.array());
        if (position + size + of_data > fileSize) {
            return null;
        }
        // 读取 checkSum 和 data
        ByteBuffer dataBuf = ByteBuffer.allocate(of_data + size);
        try {
            channel.position(position);
            channel.read(dataBuf);
        } catch (IOException e) {
            ExceptionUtils.getMessage(e);
        }
        byte[] log = dataBuf.array();
        // 校验checkSum
        int checkSumL = calCheckSum(0, Arrays.copyOfRange(log, of_data, log.length));
        int checkSumR = Parser.parseInt(Arrays.copyOfRange(log, of_checksum, of_data));
        if (checkSumL != checkSumR) {
            return null;
        }
        position += log.length;
        return log;
    }

    /**
     * 打开一个文件，首先校验txCheckSum，并移除尾部可能存在的badTail
     * 由于该条日志尚未写入完成（badTail），所以文件校验和不会包含该日志的校验和，去掉badTail就可以保证日志文件的一致性
     */
    private void checkAndRemoveTail() {
        rewind();

        int txCheck = 0;
        for (; ; ) {
            byte[] log = internNext();
            if (log == null) {
                break;
            }
            txCheck = calCheckSum(txCheck, log);
        }
        if (txCheck != txCheckSum) {
            ExceptionUtils.getMessage(new RuntimeException("Bad log file!"));
        }
        try {
            truncate(position);
        } catch (Exception e) {
            ExceptionUtils.getMessage(e);
        }
        try {
            file.seek(position);
        } catch (IOException e) {
            ExceptionUtils.getMessage(e);
        }
        rewind();
    }

    /**
     * 将data封装成日志文件规定的格式[size][checkSum][data]
     * @param data
     * @return
     */
    private byte[] wrapLog(byte[] data) {
        byte[] checkSum = Parser.intToByte(calCheckSum(0, data));
        byte[] size = Parser.intToByte(data.length);
        return Bytes.concat(size, checkSum, data);
    }

    private void updateTxCheckSum(byte[] log) {
        this.txCheckSum = calCheckSum(this.txCheckSum, log);
        try {
            channel.position(0);
            channel.write(ByteBuffer.wrap(Parser.intToByte(txCheckSum)));
            channel.force(false);
        } catch (IOException e) {
            ExceptionUtils.getMessage(e);
        }
    }


    /**
     * 向日志文件写入日志时，也是将数据包裹成日志格式，然后更新校验和，刷新缓冲区，flush disk
     * @param data
     */
    @Override
    public void log(byte[] data) {
        byte[] log = wrapLog(data);
        ByteBuffer buffer = ByteBuffer.wrap(log);
        lock.lock();
        try {
            channel.position(channel.size());
            channel.write(buffer);
        } catch (IOException e) {
            ExceptionUtils.getMessage(e);
        }finally {
            lock.unlock();
        }
        updateTxCheckSum(log);
    }

    @Override
    public void truncate(long x) throws Exception {
        lock.lock();
        try {
            channel.truncate(x);
        }finally {
            lock.unlock();
        }
    }

    @Override
    public byte[] next() {
        lock.lock();
        try {
            byte[] log = internNext();
            if (log == null) {
                return null;
            }
            return Arrays.copyOfRange(log, of_data, log.length);
        }finally {
            lock.unlock();
        }
    }

    @Override
    public void rewind() {
        this.position = 4;
    }

    @Override
    public void close() {
        try {
            channel.close();
            file.close();
        } catch (IOException e) {
            ExceptionUtils.getMessage(e);
        }
    }
}
